package com.desheng.bigdata.flink.stream.transformation

import org.apache.flink.streaming.api.scala._

/**
  * flink算子操作之split和select
  *    split是将一个datasteam根据指定的split函数拆分成为一个datastream的集--->splitstream
  *    select就是从上述拆分的流splitstream中去选择其中某一个或多个流
  *
  */
object _01TransformationForSplitAndSelectOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment


        val inputs:DataStream[String] = env.socketTextStream("bigdata01", 9999)

        val goods = inputs.map(line => {
            val fields = line.split("\\|")
            if(fields == null || fields.length != 3) {
                Goods(null, null, null)
            } else {
                Goods(fields(0), fields(1), fields(2))
            }
        }).filter(goods => goods.id != null)

        //把goods信息按照category进行分类，每一个类就形成了一个局部的数据集
        val splitStream:SplitStream[Goods] = goods.split(goods => List(goods.category))

        //想获取category为mobile的数据===
//        splitStream.select("mobile").print("mobile")
//
//        splitStream.select("sports").print("sports")

        val mobile = splitStream.select("mobile")

        val sports = splitStream.select("sports")

        mobile.union(sports).print()

        env.execute(s"${_01TransformationForSplitAndSelectOps.getClass.getSimpleName}")
    }
}
case class Goods(id: String, brand: String, category: String)
